use crate::db::Db;
use crate::models::*;
use crate::mqtt::*;
use crate::redis::*;
use anyhow::bail;
use anyhow::{Result, anyhow};
use diesel::prelude::*;
use json_patch::patch;
use num_traits::FromPrimitive;
use redis::RedisError;
use redis::{AsyncTypedCommands, HashFieldExpirationOptions, JsonAsyncCommands};
use rumqttc::v5::AsyncClient;
use rumqttc::v5::Incoming;
use rumqttc::v5::mqttbytes::QoS;
use salvo::oapi::ToSchema;
use serde::Serialize;
use std::collections::{HashMap, HashSet};
use thiserror::Error;
use tracing::{debug, info};

#[derive(Error, Debug)]
pub enum SpaceError {
    #[error("The space is inactive.")]
    Inactive,
    #[error("The space is offline.")]
    Offline,
    #[error("The space is locked.")]
    Locked,
    #[error(transparent)]
    JsonPatch(#[from] json_patch::PatchError),
    #[error(transparent)]
    SerdeJson(#[from] serde_json::Error),
    #[error(transparent)]
    Redis(#[from] RedisError),
    #[error(transparent)]
    Anyhow(#[from] anyhow::Error),
}

pub async fn list_space_active(mut redis: Redis) -> Result<HashSet<i64>> {
    let space_ids = redis
        .connection_manager
        .smembers(space_key_active())
        .await?
        .iter()
        .map(|s| s.parse().unwrap())
        .collect();
    Ok(space_ids)
}

pub async fn load_space_is_active(mut redis: Redis, space_id: i64) -> Result<bool> {
    let is_active = redis
        .connection_manager
        .sismember(space_key_active(), space_id)
        .await?;
    Ok(is_active)
}

pub async fn drop_space_active(mut redis: Redis, space_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .srem(space_key_active(), space_id)
        .await?;
    Ok(())
}

pub async fn list_space_online(mut redis: Redis) -> Result<Vec<i64>> {
    let space_ids = redis
        .connection_manager
        .hkeys(space_key_online())
        .await?
        .iter()
        .map(|s| s.parse().unwrap())
        .collect();
    Ok(space_ids)
}

pub async fn load_space_is_online(mut redis: Redis, space_id: i64) -> Result<bool> {
    let is_online = redis
        .connection_manager
        .hget(space_key_online(), &space_id)
        .await?
        .is_some();
    Ok(is_online)
}

pub async fn list_space_users(mut redis: Redis, space_id: i64) -> Result<Vec<User>> {
    let users = redis
        .connection_manager
        .hvals(space_key_users(space_id))
        .await?
        .iter()
        .fold(Vec::new(), |mut v, s| {
            if let Ok(u) = serde_json::from_str(s) {
                v.push(u);
            }
            v
        });
    Ok(users)
}

pub async fn load_space_basic(db: Db, mut redis: Redis, space_id: i64) -> Result<SpaceBasic> {
    use crate::schema::spaces;

    let key = space_key_basic(space_id);
    match redis.connection_manager.json_get(key.clone(), "$").await? {
        Some(v) => Ok(v),
        None => {
            let mut conn = db.connect()?;
            let space = spaces::table
                .select((
                    spaces::id,
                    spaces::owner_id,
                    spaces::name,
                    spaces::desp,
                    spaces::booking,
                ))
                .find(space_id)
                .get_result::<SpaceBasic>(&mut conn)?;
            let _ = redis
                .connection_manager
                .json_set::<_, _, _, ()>(key, "$", &space)
                .await?;
            Ok(space)
        }
    }
}

pub async fn drop_space_basic(mut redis: Redis, space_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .del(space_key_basic(space_id.clone()))
        .await;
    Ok(())
}

pub async fn load_space_state(db: Db, mut redis: Redis, space_id: i64) -> Result<SpaceState> {
    use crate::schema::spaces;

    let key = space_key_state(space_id);
    if !redis.connection_manager.exists(key.clone()).await? {
        let mut conn = db.connect()?;
        let space = spaces::table
            .find(space_id)
            .get_result::<Space>(&mut conn)?;
        let state = serde_json::from_value::<SpaceState>(space.state)?;
        let _ = sync_space_state(redis.clone(), space_id, &state, Some(space.state_ver)).await?;
    }
    let state = redis
        .connection_manager
        .json_get::<_, _, SpaceState>(key, "$")
        .await?;
    debug!("load space state -- {:?}", state);
    Ok(state)
}

pub async fn drop_space_state(mut redis: Redis, space_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .del(space_key_state(space_id.clone()))
        .await;
    let _ = redis
        .connection_manager
        .del(space_key_state_ver(space_id.clone()))
        .await;
    Ok(())
}

pub async fn list_space_perms(mut redis: Redis, space_id: i64) -> Result<Vec<(i64, usize)>> {
    let perms = redis
        .connection_manager
        .zrangebyscore_withscores(space_key_perms(space_id), 0_usize, 2_usize)
        .await?
        .iter()
        .map(|(s, u)| (s.parse().unwrap(), u.to_owned()))
        .collect();
    Ok(perms)
}

pub async fn drop_space_perms(mut redis: Redis, space_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .del(space_key_perms(space_id.clone()))
        .await;
    Ok(())
}

pub async fn list_space_pings(mut redis: Redis, space_id: i64) -> Result<Vec<i64>> {
    let pings = redis
        .connection_manager
        .hkeys(space_key_pings(space_id))
        .await?
        .iter()
        .map(|s| s.parse().unwrap())
        .collect();
    Ok(pings)
}

pub async fn drop_space_pings(mut redis: Redis, space_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .del(space_key_pings(space_id.clone()))
        .await;
    Ok(())
}

pub async fn sync_space_active(mut redis: Redis, space_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .sadd(space_key_active(), space_id)
        .await?;
    Ok(())
}

pub async fn sync_space_online(mut redis: Redis, space_id: i64) -> Result<()> {
    let options = HashFieldExpirationOptions::default().set_expiration(redis::SetExpiry::EX(600));
    let _ = redis
        .connection_manager
        .hset_ex(space_key_online(), &options, &[(space_id, "")])
        .await?;
    Ok(())
}

pub async fn sync_space_state(
    mut redis: Redis,
    space_id: i64,
    state: &SpaceState,
    version: Option<i64>,
) -> Result<i64> {
    let key = space_key_state_ver(space_id);
    let ver_now = match version {
        Some(ver) => {
            let _ = redis.connection_manager.set(key.clone(), ver).await?;
            ver
        }
        None => {
            let ver = redis.connection_manager.incr(key, 1_i64).await?;
            i64::from_isize(ver).ok_or(anyhow!("cannot convert 'version' from isize to i64."))?
        }
    };
    let _ = redis
        .connection_manager
        .json_set::<_, _, _, ()>(
            space_key_state(space_id),
            "$",
            &serde_json::to_value(state)?,
        )
        .await?;
    Ok(ver_now)
}

pub async fn load_space_user_basic(
    db: &Db,
    mut redis: Redis,
    space_id: i64,
    user_id: i64,
) -> Result<User> {
    use crate::schema::users;

    let key = space_key_users(space_id);
    let user = match redis.connection_manager.hget(key.clone(), user_id).await? {
        Some(s) => serde_json::from_str::<User>(&s)?,
        None => {
            let mut conn = db.connect()?;
            let user = users::table.find(user_id).get_result::<User>(&mut conn)?;
            let options =
                HashFieldExpirationOptions::default().set_expiration(redis::SetExpiry::EX(86400));
            let _ = redis
                .connection_manager
                .hset_ex(
                    key.clone(),
                    &options,
                    &[(user_id, serde_json::to_string(&user)?)],
                )
                .await?;
            user
        }
    };
    Ok(user)
}

pub async fn drop_space_user_basic(mut redis: Redis, space_id: i64, user_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .hdel(space_key_users(space_id), user_id)
        .await?;
    Ok(())
}

pub enum PermRole {
    Teacher,
    Student,
    Watcher,
}

impl PermRole {
    fn to_int(&self) -> usize {
        match self {
            PermRole::Teacher => 2_usize,
            PermRole::Student => 1_usize,
            PermRole::Watcher => 0_usize,
        }
    }
}

pub async fn sync_space_user_perm(
    mut redis: Redis,
    client: &AsyncClient,
    space: SpaceBasic,
    user_id: i64,
    role: PermRole,
) -> Result<()> {
    let _ = redis
        .connection_manager
        .zadd(space_key_perms(space.id), user_id, role.to_int())
        .await?;
    let _ = publish_perms(client, redis, space.id).await?;
    Ok(())
}

pub async fn drop_space_user_perm(
    mut redis: Redis,
    client: &AsyncClient,
    space: SpaceBasic,
    user_id: i64,
) -> Result<()> {
    let _ = redis
        .connection_manager
        .zrem(space_key_perms(space.id), user_id)
        .await?;
    let _ = publish_perms(client, redis, space.id).await?;
    Ok(())
}

pub async fn sync_space_user_ping(mut redis: Redis, space_id: i64, user_id: i64) -> Result<()> {
    let options = HashFieldExpirationOptions::default().set_expiration(redis::SetExpiry::EX(60));
    let _ = redis
        .connection_manager
        .hset_ex(space_key_pings(space_id), &options, &[(user_id, "")])
        .await?;
    Ok(())
}

pub async fn drop_space_user_ping(mut redis: Redis, space_id: i64, user_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .hdel(space_key_pings(space_id), user_id)
        .await?;
    Ok(())
}

pub async fn drop_space_unused(db: Db, redis: Redis) -> Result<()> {
    use crate::schema::spaces;

    let active_ids: HashSet<i64> = list_space_active(redis.clone()).await?;
    let online_ids: HashSet<i64> = HashSet::from_iter(list_space_online(redis.clone()).await?);
    for space_id in active_ids.difference(&online_ids) {
        if !load_space_is_online(redis.clone(), space_id.to_owned()).await? {
            info!("drop space unused -- {}", space_id);
            let state = load_space_state(db.clone(), redis.clone(), space_id.to_owned()).await?;
            let mut conn = db.connect()?;
            let _ = diesel::update(spaces::table.filter(spaces::id.eq(space_id.clone())))
                .set(spaces::state.eq(serde_json::to_value(state)?))
                .execute(&mut conn)?;
            let _ = drop_space_basic(redis.clone(), space_id.to_owned()).await?;
            let _ = drop_space_state(redis.clone(), space_id.to_owned()).await?;
            let _ = drop_space_perms(redis.clone(), space_id.to_owned()).await?;
            let _ = drop_space_pings(redis.clone(), space_id.to_owned()).await?;
            let _ = drop_space_active(redis.clone(), space_id.to_owned()).await?;
        }
    }
    Ok(())
}

#[derive(Serialize, ToSchema)]
pub struct UserPlus {
    pub id: i64,
    pub username: String,
    pub avatar: String,
    pub perm: usize,
    pub online: bool,
}

pub async fn publish_perms(client: &AsyncClient, redis: Redis, space_id: i64) -> Result<()> {
    let basic_users = list_space_users(redis.clone(), space_id).await?;
    let perms = list_space_perms(redis.clone(), space_id).await?;
    let perm_map: HashMap<i64, usize> = HashMap::from_iter(perms);
    let pings = list_space_pings(redis, space_id).await?;
    let users = basic_users
        .iter()
        .map(|user| UserPlus {
            id: user.id,
            username: user.username.to_owned(),
            avatar: user.avatar.to_owned(),
            perm: perm_map
                .get(&user.id)
                .map(|v| v.to_owned())
                .unwrap_or_default(),
            online: pings.contains(&user.id),
        })
        .collect::<Vec<UserPlus>>();
    let payload = serde_json::to_string(&users)?;
    let _ = client
        .publish(topic_space_perms(space_id), QoS::ExactlyOnce, true, payload)
        .await?;
    Ok(())
}

pub async fn publish_patch(
    client: &AsyncClient,
    space_id: i64,
    payload: &PayloadPatch,
) -> Result<()> {
    let _ = client
        .publish(
            topic_space_patch(space_id),
            QoS::ExactlyOnce,
            false,
            serde_json::to_string(payload)?,
        )
        .await?;
    Ok(())
}

pub async fn publish_pings(client: &AsyncClient, redis: Redis) -> Result<()> {
    let space_ids = list_space_online(redis.clone()).await?;
    for id in space_ids {
        debug!("publish pings for space [{}]", id);
        let pings = list_space_pings(redis.clone(), id).await?;
        let payload = serde_json::to_string(&pings)?;
        client
            .publish(topic_state_pings(id), QoS::AtMostOnce, false, payload)
            .await?;
    }
    Ok(())
}

pub async fn proc_incoming(
    incoming: Incoming,
    db: Db,
    redis: Redis,
    client: &AsyncClient,
) -> Result<()> {
    let _: Result<()> = match incoming {
        Incoming::Publish(v) => {
            let topic = String::from_utf8(v.topic.into())?;
            info!("incoming topic -- [{}]", topic);
            let user_props: HashMap<String, String> = HashMap::from_iter(
                v.properties
                    .ok_or(anyhow!("'user_properties' not found"))?
                    .user_properties,
            );
            let space_id: i64 = user_props
                .get("space_id")
                .ok_or(anyhow!("'space_id' not found in 'user_properties'"))?
                .parse()?;
            let oper_id: i64 = user_props
                .get("oper_id")
                .ok_or(anyhow!("'oper_id' not found in 'user_properties'"))?
                .parse()?;
            if topic == topic_pings() {
                let _ = proc_topic_pings(space_id, oper_id, redis.clone()).await?;
            }
            if topic == topic_perms() {
                let payload = serde_json::from_str::<PayloadPerm>(
                    String::from_utf8(v.payload.clone().into())?.as_str(),
                )?;
                let _ = proc_topic_perms(
                    space_id,
                    oper_id,
                    payload,
                    db.clone(),
                    redis.clone(),
                    client,
                )
                .await?;
            }
            if topic == topic_patch() {
                let payload = serde_json::from_str::<PayloadPatch>(
                    String::from_utf8(v.payload.into())?.as_str(),
                )?;
                let _ = proc_topic_patch(space_id, oper_id, payload, db, redis, client).await?;
            }
            Ok(())
        }
        _ => Ok(()),
    };
    Ok(())
}

pub async fn proc_topic_pings(space_id: i64, oper_id: i64, redis: Redis) -> Result<()> {
    info!("proc pings from space [{}], user [{}]", space_id, oper_id);
    let _ = sync_space_online(redis.clone(), space_id).await?;
    let _ = sync_space_user_ping(redis, space_id, oper_id);
    Ok(())
}

pub async fn proc_topic_perms(
    space_id: i64,
    oper_id: i64,
    payload: PayloadPerm,
    db: Db,
    redis: Redis,
    client: &AsyncClient,
) -> Result<()> {
    info!("proc perms from space [{}], user [{}]", space_id, oper_id);
    let space = load_space_basic(db, redis.clone(), space_id).await?;
    let role = match payload.perm {
        true => PermRole::Student,
        false => PermRole::Watcher,
    };
    if space.owner_id == oper_id && oper_id != payload.user_id {
        let _ = sync_space_user_perm(redis, client, space, payload.user_id, role).await?;
    }
    Ok(())
}

pub async fn proc_topic_patch(
    space_id: i64,
    oper_id: i64,
    payload: PayloadPatch,
    db: Db,
    redis: Redis,
    client: &AsyncClient,
) -> Result<()> {
    info!("proc patch from space [{}], user [{}]", space_id, oper_id);
    let perms: HashMap<i64, usize> =
        HashMap::from_iter(list_space_perms(redis.clone(), space_id).await?);
    if let Some(perm) = perms.get(&oper_id) {
        if *perm > 0_usize {
            let version = patch_state(space_id, db, redis, &payload).await?;
            let _ = publish_patch(
                client,
                space_id,
                &PayloadPatch {
                    version: Some(version),
                    ..payload
                },
            )
            .await?;
        }
    }
    Ok(())
}

pub async fn space_state_lock(mut redis: Redis, space_id: i64) -> Result<bool> {
    let space_ids = list_space_active(redis.clone()).await?;
    if space_ids.contains(&space_id) {
        let options = HashFieldExpirationOptions::default()
            .set_expiration(redis::SetExpiry::EX(10))
            .set_existence_check(redis::FieldExistenceCheck::FNX);
        let success = redis
            .connection_manager
            .hset_ex(space_key_lock(), &options, &[(space_id, "")])
            .await?;
        Ok(success)
    } else {
        bail!("space is not active")
    }
}

pub async fn space_state_unlock(mut redis: Redis, space_id: i64) -> Result<()> {
    let _ = redis
        .connection_manager
        .hdel(space_key_lock(), space_id)
        .await?;
    Ok(())
}

pub async fn patch_state(
    space_id: i64,
    db: Db,
    redis: Redis,
    payload: &PayloadPatch,
) -> Result<i64, SpaceError> {
    if space_state_lock(redis.clone(), space_id).await? {
        let res = async {
            let state_old = load_space_state(db, redis.clone(), space_id.clone()).await?;
            let mut state_v = serde_json::to_value(state_old)?;
            patch(&mut state_v, &payload.patch)?;
            let state_new = serde_json::from_value::<SpaceState>(state_v)?;
            let version = sync_space_state(redis.clone(), space_id, &state_new, None).await?;
            Ok(version)
        }
        .await;
        let _ = space_state_unlock(redis, space_id);
        res
    } else {
        Err(SpaceError::Locked)
    }
}
